package com.atguigu.edu.realtime.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwsTrafficBaseLogSplit {
    public static void main(String[] args) {
        // TODO 1. 基本环节准备
        // 1.1 指定流处理环节
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(4);
        // TODO 2.检查点相关的设置(略)
        // TODO 3.从kafka主题中读取数据
        // 3.1 声明消费的主题以及消费组
        String topic = "topic_log";
        String groupId = "dws_traffic_log_split_group";
        // 3.2 创建消费组对象  【flink在消费kafka数据时，偏移量由Flink维护】
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据 封装为流
        DataStreamSource<String> kafkaDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");
        //kafkaDs.print("输入数据");
        // TODO 4.对读取的数据进行类型转换并ETC jsonStr->jsonObj 将脏数据放到侧输出流中
        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag"){};//本类会出现泛型擦除，所以用匿名内部子类
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaDs.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyTag, "dirtyTag");
                }
            }
        });
        SideOutputDataStream<String> dirtyDs = jsonObjDs.getSideOutput(dirtyTag);//捕获测流
        // TODO 5.将侧输出流中的脏数据写到kafka主题中
        dirtyDs.sinkTo(MyKafkaUtil.getKafkaSink("dirty_data"));
        // TODO 6.按照设备的mid进行分组  【因为要做状态恢复】
        KeyedStream<JSONObject, String> keyedDs = jsonObjDs.keyBy(js -> js.getJSONObject("common").getString("mid"));
        // TODO 7.使用Flink的状态编程，对新老访客状态恢复
        SingleOutputStreamOperator<JSONObject> fixedDs = keyedDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            //用于储存老访客最早访问时间的状态
            //注意：不要在声明状态的时候对其进行初始化，因为这个时候还获取不到运行时上下文
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                //状态中的首次访问日期
                String value = valueState.value();
                String date = DateFormatUtil.toDate(jsonObj.getLong("ts"));
                String isNew = jsonObj.getJSONObject("common").getString("is_new");
                if (StringUtils.isNotEmpty(value)) {
                    if (isNew.equals("1")) { //可能用户本地缓存被清除掉，将老访客误认为新访客
                        if (!value.equals(date)) {//当前日期与注册日期不一致，修复新客户的状态
                            isNew = "0";
                            jsonObj.getJSONObject("common").put("is_new", isNew);
                        }
                    } else {
                        if (isNew.equals("0")) {
                            valueState.update(DateFormatUtil.toDate(jsonObj.getLong("ts") - 24 * 60 * 60 * 1000));
                        } else {
                            valueState.update(date);
                        }
                    }
                }
                out.collect(jsonObj);
            }
        });
        fixedDs.print("状态恢复");
        // TODO 8.分流
        //  错误日志-错误侧输出流
        //  启动日志-启动侧输出流
        //  曝光日志-曝光侧输出流
        //  动作日志-动作侧输出流
        //  页面日志-主流
        // 8.1 定义侧输出流标签
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideoTag"){};
        // 8.2 分流
        SingleOutputStreamOperator<String> pageDs = fixedDs.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                JSONObject err = value.getJSONObject("err");
                //~~~错误日志~~~
                if (err != null) {
                    ctx.output(errTag, value.toJSONString());//输出整条数据，方便获取机型，时间，地址等数据
                    value.remove("err");//将错误的相关信息删除掉，剩下的往后传递
                }
                //~~~启动日志~~~
                JSONObject start = value.getJSONObject("start");
                JSONObject appVideo = value.getJSONObject("appVideo");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());//输出整条数据，方便获取机型，时间，地址等数据
                } else if(appVideo != null){
                    ctx.output(actionTag,value.toJSONString());
                }
                else{//要不然启动日志，要不然页面浏览
                    // 公共数据
                    JSONObject commonJsonObj = value.getJSONObject("common");
                    JSONObject pageJsonObj = value.getJSONObject("page");
                    Long ts = value.getLong("ts");
                    //~~~曝光日志~~~
                    JSONArray displayArray = value.getJSONArray("displays");// todo:注意s
                    if (displayArray != null && displayArray.size() > 0) {
                        for (int i = 0; i < displayArray.size(); i++) {
                            JSONObject jsonObject = displayArray.getJSONObject(i);
                            JSONObject object = new JSONObject();
                            object.put("common", commonJsonObj);
                            object.put("page", pageJsonObj);
                            object.put("display", jsonObject);
                            object.put("ts", ts);
                            //曝光数据输出到曝光侧输出流中
                            ctx.output(displayTag, object.toJSONString());
                        }
                        value.remove("displays");
                    }
                    //~~~动作日志~~~
                    JSONArray actionsArray = value.getJSONArray("actions");// todo:注意s
                    if (actionsArray != null && actionsArray.size() > 0) {
                        for (int i = 0; i < actionsArray.size(); i++) {
                            JSONObject actionJsonObj = actionsArray.getJSONObject(i);
                            JSONObject newActionJsonObj = new JSONObject();
                            newActionJsonObj.put("common", commonJsonObj);
                            newActionJsonObj.put("page", pageJsonObj);
                            newActionJsonObj.put("display", actionJsonObj);
                            //newActionJsonObj.put("ts",ts);动作里维护了操作时间，ts是进入时间，不匹配，也不用给，自己有
                            //将动作日志输出到动作侧输出流
                            ctx.output(actionTag, newActionJsonObj.toJSONString());
                        }
                        value.remove("actions");// todo:注意s
                    }
                    //~~~页面日志~~~
                    out.collect(value.toJSONString());//还有common，ts，page
                }
            }
        });
        // TODO 9.将不同的流的数据写到不同的kafka主题中
        //捕获侧流
        SideOutputDataStream<String> errDS = pageDs.getSideOutput(errTag);
        SideOutputDataStream<String> startDS = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> displayDS = pageDs.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDS = pageDs.getSideOutput(actionTag);
        SideOutputDataStream<String> appVideoDs = pageDs.getSideOutput(appVideoTag);
        pageDs.print(">>>>");
        errDS.print("###");
        startDS.print("$$$");
        displayDS.print("&&&");
        actionDS.print("***");
        appVideoDs.print("@@@");
        pageDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_page_log"));
        errDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_err_log"));
        startDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_start_log"));
        displayDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_display_log"));
        actionDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_action_log"));
        appVideoDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_appVideo_log"));
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
